package com.example.springbootkafka.controller;

import com.alibaba.fastjson.JSONObject;
import com.example.springbootkafka.config.KafkaConfig;
import com.example.springbootkafka.dto.TtlMsgDto;
import com.example.springbootkafka.service.ProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.connector.Response;
import org.apache.kafka.streams.KafkaStreams;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.stream.IntStream;

@Slf4j
@RestController
@RequestMapping("/test")
public class Controller {


    @Autowired
    ProducerService producerService;


    @GetMapping("/sendMsg")
    public Integer sendMsg(@RequestParam("msg") String msg) {
        //producerService.sendMessage("testTopic1", "key", msg);
        long delayedSeconds = 12L;
        TtlMsgDto dto = new TtlMsgDto();
        dto.setData(msg);
        dto.setExpireTime(LocalDateTime.now().plusSeconds(delayedSeconds));
        dto.setMessageId(UUID.randomUUID().toString());
        producerService.sendTtlMessage("my-test-delayed-topic", JSONObject.toJSONString(dto), delayedSeconds * 1000L);
        return Response.SC_OK;
    }

    @Resource
    private StreamsBuilderFactoryBean factoryBean;

    @GetMapping("/sendStreamMsg")
    public Integer sendStreamMsg(@RequestParam("msg") String msg) {
        IntStream.range(1,1000).boxed().forEach(x->{
            long delayedSeconds = 5L;
            TtlMsgDto dto = new TtlMsgDto();
            dto.setData(msg);
            dto.setExpireTime(LocalDateTime.now().plusSeconds(delayedSeconds));
            dto.setMessageId(UUID.randomUUID().toString());
            producerService.sendTtlMessage(KafkaConfig.STREAM_IN_TOPIC, "key", JSONObject.toJSONString(dto), delayedSeconds * 1000L);

        });


        return Response.SC_OK;
    }


}
